package org.ilc.socketserver;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.filter.executor.OrderedThreadPoolExecutor;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.ilc.exception.DisposeException;
import org.ilc.push.ISessionListener;
import org.ilc.socketserver.codec.DefaultProtocolCodecFactory;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author malongbo
 */
public class Acceptor {
    private Log log = LogFactory.getLog ("PushServer");
    private NioSocketAcceptor acceptor = null;
    private ThreadPoolExecutor threadPool = null;
    private int corePoolSize = 5;
    private int maxPoolSize = 50;
    //主服务监听的端口可以重用
    private boolean reuseAddress = true;
    private int receiveBufferSize = 512;
    private int sendBufferSize = 2048;
    //设置为非延迟发送，为true则不组装成大包发送，收到东西马上发出
    private boolean tcpNoDelay = true;
    //心跳检测间隔时间
    private int idleTime = 180;
    private int port = 8889;
    private SessionMapping mapping;
    private ISessionListener sessionListener;

    public Acceptor(){
        mapping = new SessionMapping();
    }

    /**
     * 服务器是否运行
     * @return
     */
    public boolean isActive(){
        return (null != this.acceptor && this.acceptor.isActive());
    }


    public void assertDispose(){
        if(null != this.acceptor && !this.acceptor.isActive()){
            throw new DisposeException ("Push Server is Closed");
        }
    }

    public Long getSessionId(String user){
        return this.mapping.getId(user);
    }

    public void removeMapping(String user){
        this.mapping.remove(user);
    }

    public IoSession getSession(String user){
        if(mapping.isExist(user)){
            Long id = mapping.getId(user);
            return getSession(id);
        }
        return null;
    }

    /**
     * 根据session id 取session
     * @param id
     * @return
     */
    public IoSession getSession(Long id){
        return this.acceptor.getManagedSessions().get(id);
    }

    public ISessionListener getSessionListener() {
        return sessionListener;
    }

    public void setSessionListener(ISessionListener sessionListener) {
        this.sessionListener = sessionListener;
    }

    public boolean isExist(String user){
        return this.mapping.isExist(user);
    }

    public void addMapping(String user,Long sessionId){
        this.mapping.register(user, sessionId);
    }

    /**
     * 取所有的session
     * @return
     */
    public Collection<IoSession> getAllSessions(){
        return this.acceptor.getManagedSessions().values();
    }

    public void start () {
        if(null != acceptor && acceptor.isActive()){
            if(log.isWarnEnabled()){
                log.warn("server already startup ");
            }
            return;
        }
        if (log.isInfoEnabled()) {
            log.info("server start...");
        }

        try {
            acceptor = new NioSocketAcceptor(Runtime.getRuntime ().availableProcessors () + 1);
            /*threadPool = new ThreadPoolExecutor(
                    this.corePoolSize, this.maxPoolSize, 0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable> ());*/
            OrderedThreadPoolExecutor threadPool = new OrderedThreadPoolExecutor (
                    this.corePoolSize, this.maxPoolSize);

            acceptor.getFilterChain ().addLast ("codec", new ProtocolCodecFilter (new DefaultProtocolCodecFactory()));
            acceptor.getFilterChain().addLast("exector", new ExecutorFilter (threadPool));
            acceptor.setReuseAddress(this.reuseAddress);// 设置的是主服务监听的端口可以重用
            acceptor.getSessionConfig().setReceiveBufferSize(
                    this.receiveBufferSize);// 设置输入缓冲区的大小
            acceptor.getSessionConfig().setSendBufferSize(
                    this.sendBufferSize);// 设置输出缓冲区的大小
            // 设置为非延迟发送，为true则不组装成大包发送，收到东西马上发出
            acceptor.getSessionConfig().setTcpNoDelay(this.tcpNoDelay);
            acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE,
                    this.idleTime);

            // //设置主服务监听端口的监听队列的最大值为100，如果当前已经有100个连接，再新的连接来将被服务 器拒绝
            // acceptor.setBacklog(100);

            acceptor.setHandler(new SessionHandler (this));
            acceptor.bind(new InetSocketAddress (this.port));
            if (log.isInfoEnabled()) {
                log.info(new StringBuffer("start on :")
                        .append (this.port));
                log.info(new StringBuffer("core pool size :")
                        .append (this.corePoolSize));
                log.info(new StringBuffer("max pool size :")
                        .append (this.maxPoolSize));
                log.info(new StringBuffer("reuseaddress :")
                        .append (this.reuseAddress));
                log.info(new StringBuffer("recevie buff size :")
                        .append (this.receiveBufferSize));
                log.info(new StringBuffer("send buff size :")
                        .append (this.sendBufferSize));
                log.info(new StringBuffer("tcpnodeplay :")
                        .append (this.tcpNoDelay));
                log.info(new StringBuffer("idle time :")
                        .append (this.idleTime));
            }
        } catch (Exception e) {
            log.error("start error", e);
        }
    }

    public void stop () {
        if (log.isInfoEnabled()) {
            log.info("server is stop ...");
        }
        try {
            this.threadPool.shutdown();
            acceptor.unbind();
            acceptor.dispose();
            threadPool=null;
            acceptor = null;
        } catch (Exception e) {
            // TODO: handle exception
        }
    }

    public void restart() {
        if (log.isInfoEnabled()) {
            log.info("restart...");
        }
        stop();
        start();
    }

    public int getCorePoolSize() {
        return corePoolSize;
    }

    public void setCorePoolSize(int corePoolSize) {
        this.corePoolSize = corePoolSize;
    }

    public int getMaxPoolSize() {
        return maxPoolSize;
    }

    public void setMaxPoolSize(int maxPoolSize) {
        this.maxPoolSize = maxPoolSize;
    }

    public boolean isReuseAddress() {
        return reuseAddress;
    }

    public void setReuseAddress(boolean reuseAddress) {
        this.reuseAddress = reuseAddress;
    }

    public int getReceiveBufferSize() {
        return receiveBufferSize;
    }

    public void setReceiveBufferSize(int receiveBufferSize) {
        this.receiveBufferSize = receiveBufferSize;
    }

    public int getSendBufferSize() {
        return sendBufferSize;
    }

    public void setSendBufferSize(int sendBufferSize) {
        this.sendBufferSize = sendBufferSize;
    }

    public boolean isTcpNoDelay() {
        return tcpNoDelay;
    }

    public void setTcpNoDelay(boolean tcpNoDelay) {
        this.tcpNoDelay = tcpNoDelay;
    }

    public int getIdleTime() {
        return idleTime;
    }

    public void setIdleTime(int idleTime) {
        this.idleTime = idleTime;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public SessionMapping getMapping() {
        return mapping;
    }

}
